package com.lagou.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

/**
 *
 * 需求2： 删除好友操作实现（好友关系双向，一方删除好友，另一方也会被迫删除好友）
 *
 * 例如：uid1用户执行删除uid2这个好友，则uid2的好友列表中也必须删除uid1
 *
 */

public class MyProcessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取表对象
        HTableInterface relations = e.getEnvironment().getTable(TableName.valueOf("relations"));
        //获取删除列族队列的集合
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        //循环遍历获得cells对象
        for (Map.Entry<byte[], List<Cell>> entry : entries) {
            System.out.println(Bytes.toString(entry.getKey()));
            List<Cell> cells = entry.getValue();
            for (Cell cell : cells) {
                //获取每个cell的rowkey和列（qualifier）
                byte[] row = CellUtil.cloneRow(cell);
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                //判断用户“qualifier”是否删除好友用户“row”
                Get get = new Get(qualifier);
                get.addColumn(Bytes.toBytes("friends"), row);
                boolean exists = relations.exists(get);
                //存在用户则进行删除，否则跳过
                if (exists) {
                    Delete delete1 = new Delete(qualifier);
                    delete1.addColumn(Bytes.toBytes("friends"), row);
                    relations.delete(delete1);
                }
            }
        }
    }
}

